package com.atguigu.bigdata.gmall.realtime.app.dwd;

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class Dwd_02_DwdTradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_02_DwdTradePayDetailSuc().init(
                2003,
                2,
                "Dwd_02_DwdTradePayDetailSuc"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        // 1. 读取ods_db
        readOdsDb(tEnv, "Dwd_02_DwdTradePayDetailSuc");

        // 2. 从 ods_db 过滤出 支付成功数据
        Table paymentInfo = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['order_id'] order_id, " +
                "data['payment_status'] payment_status," +
                "data['callback_time'] callback_time, " +
                "ts, " +
                "pt " +
                "from ods_db " +
                "where `database`='gmall2022' " +
                "and `table`='payment_info' " +
                "and `type`='insert' " +
                "and `data`['payment_status']='1602' ");
        tEnv.createTemporaryView("payment_info", paymentInfo);


        // 4. 读取下单详情表  (Dwd_03_DwdTradeOrderDetail 写入的数据)
        tEnv.executeSql(" create table dwd_trade_order_detail(" +
                "  id string," +
                "  course_id string," +
                "  course_name string," +
                "  order_id string," +
                "  user_id string," +
                "  province_id string," +
                "  origin_amount string," +
                "  final_amount string," +
                "  order_status string," +
                "  ts bigint," +
                "  row_op_ts timestamp_ltz(3)" +
                "  )" + SQLUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_02_DwdTradePayDetailSuc"));
//        tEnv.sqlQuery("select * from dwd_trade_order_detail").execute().print();

        // 5. 两张表进行 join
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        Table result = tEnv.sqlQuery(" select " +
                " pi.id, " +
                " pi.order_id," +
                " pi.payment_status," +
                " pi.callback_time," +
                " od.user_id," +
                " od.origin_amount," +
                " od.final_amount," +
                " od.ts," +
                " od.row_op_ts row_op_ts" +
                " from dwd_trade_order_detail od " +
                " join  payment_info pi on od.order_id=pi.order_id ");
//        result.execute().print();

        // 5.写入到kafka
        tEnv.executeSql(" create table dwd_trade_pay_detail_suc( " +
                " id string, " +
                " order_id string, " +
                " payment_status string, " +
                " callback_time string, " +
                " user_id string, " +
                " origin_amount string, " +
                " final_amount string, " +
                " ts bigint, " +
                " row_op_ts timestamp_ltz(3)" +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));
        result.executeInsert("dwd_trade_pay_detail_suc");
    }




}

